package com.ku.pkg.service.mq;

import cn.hutool.json.JSONUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.core.OrderComparator;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @program: rocketMQKill
 * @description: 订单服务发过来的顺序消息
 * @author: JokerKu
 * @create: 2020-05-08 11:35
 **/
@Component
public class OrderlyConsumer {

    private DefaultMQPushConsumer consumer;

    private static final String NAME_SERVER = "192.168.43.10:9876;192.168.43.9:9876";

    private static final String CONSUMER_GROUP_NAME = "orderly_consumer_group";

    public static final String PKG_TOPIC = "pkg_topic";
    public static final String PKG_TAGS = "pkg_tag";

    private OrderlyConsumer() throws MQClientException {
        this.consumer = new DefaultMQPushConsumer(CONSUMER_GROUP_NAME);
//        this.consumer.setC
        this.consumer.setNamesrvAddr(NAME_SERVER);
        this.consumer.subscribe(PKG_TOPIC, PKG_TAGS);
        this.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        this.consumer.registerMessageListener(new MessageListenerOrderlyImpl());
        this.consumer.start();
    }

    //MessageListenerOrderly 单线程的监听一个队列
    class MessageListenerOrderlyImpl implements MessageListenerOrderly {

        private final Random random = new Random();

        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext consumeOrderlyContext) {
            for (MessageExt msg : msgs) {
                try {
                    System.out.println(msg + "----" + new String(msg.getBody()));
                    String topic = msg.getTopic();
                    String keys = msg.getKeys();
                    String tags = msg.getTags();
                    //获取消息
                    String msgStr = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                   // System.err.println("平台账户：" + "topic:" + topic + "---keys:" + keys + "---tags:" + tags + "---msg:" + msg);
                    Map<String, Object> params = JSONUtil.toBean(msgStr, Map.class);
                    String userId = (String) params.get("userId");
                    String orderId = (String) params.get("orderId");
                    String desc = (String) params.get("desc");
                    System.out.println("userId:"+userId+"---orderId:"+orderId+"---desc:"+desc);
                    //模拟业务逻辑处理
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException | UnsupportedEncodingException e) {
                    e.printStackTrace();
                    //出现异常
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }
            return ConsumeOrderlyStatus.SUCCESS;
        }
    }
}
